package com.bw.gmall.realtime.dwd.db.split.app;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TableProcessDwd;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import com.bw.gmall.realtime.common.util.FlinkSourceUtil;
import com.bw.gmall.realtime.dwd.db.split.func.DwdProcessFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Arrays;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dwd.db.split.app
 * @className: DwdBaseDb
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/2/27 13:35
 * @version: 1.0
 */

public class DwdBaseDb extends BaseApp {
    public static void main(String[] args) {
        new DwdBaseDb().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_BASE_DB, 4, 10019);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        //etl处理
        SingleOutputStreamOperator<JSONObject> etl = getEtl(dataStreamSource).setParallelism(1);
        //读取配置表
        DataStreamSource<String> cdcStream = env.fromSource(FlinkSourceUtil.getMysqlSource(Constant.PROCESS_DATABASE, Constant.PROCESS_DWD_TABLE_NAME), WatermarkStrategy.noWatermarks(), "cdc_stream");
//        cdcStream.print();
        SingleOutputStreamOperator<TableProcessDwd> tableProcessDwdSingleOutputStreamOperator = tableProcessDwdSingleOutputStreamOperator(cdcStream);
        //广播流
        MapStateDescriptor<String, TableProcessDwd> broadcastState = new MapStateDescriptor<>("broadcast_state", String.class, TableProcessDwd.class);
        BroadcastStream<TableProcessDwd> broadcast = tableProcessDwdSingleOutputStreamOperator.broadcast(broadcastState);

        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> processBroadStream = etl.connect(broadcast).process(new DwdProcessFunction(broadcastState));


//        processBroadStream.print();
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> map = getMap(processBroadStream);
        map.sinkTo(FlinkSinkUtil.getDwdKafkaSink());
    }

    private static SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> getMap(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> processBroadStream) {
        return processBroadStream.map(new MapFunction<Tuple2<JSONObject, TableProcessDwd>, Tuple2<JSONObject, TableProcessDwd>>() {

            @Override
            public Tuple2<JSONObject, TableProcessDwd> map(Tuple2<JSONObject, TableProcessDwd> jsonObjectTableProcessDwdTuple2) throws Exception {
                JSONObject jsonObject = jsonObjectTableProcessDwdTuple2.f0;
                JSONObject data = jsonObject.getJSONObject("data");
                ArrayList<String> list = new ArrayList<>(Arrays.asList(jsonObjectTableProcessDwdTuple2.f1.getSinkColumns().split(",")));
                data.keySet().removeIf(key -> !list.contains(key));
                return jsonObjectTableProcessDwdTuple2;
            }

        });
    }

    private static SingleOutputStreamOperator<TableProcessDwd> tableProcessDwdSingleOutputStreamOperator(DataStreamSource<String> cdcStream) {
        return cdcStream.flatMap(new RichFlatMapFunction<String, TableProcessDwd>() {
            @Override
            public void flatMap(String s, Collector<TableProcessDwd> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String op = jsonObject.getString("op");
                System.out.println(jsonObject);
                TableProcessDwd tableProcessDwd;
                if ("d".equals(op)) {
                    String before = jsonObject.getString("before");
                    tableProcessDwd = JSONObject.parseObject(before, TableProcessDwd.class);
                } else {
                    String after = jsonObject.getString("after");
                    tableProcessDwd = JSONObject.parseObject(after, TableProcessDwd.class);
                }
                tableProcessDwd.setOp(op);
                collector.collect(tableProcessDwd);
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getEtl(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    collector.collect(jsonObject);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}
